package com.yanxu;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 */
public class Api03_Source_Kafka {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子

        //从kafka 中接收源算子
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                //kafka 服务的地址
                .setBootstrapServers("192.168.1.30:9092")
                //kafka 主题
                .setTopics("clicks")
                //kafka 组
                .setGroupId("consumer-group")
                //kafka 消费的offset
                .setStartingOffsets(OffsetsInitializer.latest())  //latest 代表从上次读取的偏移量读取
                //.setStartingOffsets(OffsetsInitializer.earliest())  //earliest 代表从头开始读取
                //序列化
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        //fromSource ---> kafka
        DataStreamSource<String> streamSource = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");
        streamSource.print("kafka source >>>");


        //对环境进行执行。
        env.execute();
    }
}